package main

import (
	"context"
	"fmt"

	sarama "github.com/Shopify/sarama"
)

type consumerGroupHandler struct {
	name string
}

func (consumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
func (consumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
func (h consumerGroupHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {

	for msg := range claim.Messages() {
		fmt.Println(msg.Offset)
		sess.MarkMessage(msg, "")
		sess.Commit()
	}
	return nil
}

func main() {

	fmt.Println("consumer_test")

	config := sarama.NewConfig()
	config.Consumer.Return.Errors = false
	config.Version = sarama.V0_11_0_2
	config.Consumer.Offsets.AutoCommit.Enable = false
	config.Consumer.Offsets.Initial = sarama.OffsetNewest

	group, err := sarama.NewConsumerGroup([]string{"120.26.95.197:9092"}, "t", config)
	//sarama.NewConsumerGroupFromClient()
	if err != nil {
		panic(err)
	}
	defer group.Close()

	for {
		handler := consumerGroupHandler{name: "sera"}
		err := group.Consume(context.Background(), []string{"test"}, handler)
		if err != nil {
			fmt.Println(err.Error())
		}
	}
}

// package main

// import (
// 	"fmt"

// 	"github.com/Shopify/sarama"
// )

// //https://www.jianshu.com/p/3d4655cd7054
// func main() {
// 	config := sarama.NewConfig()
// 	// config.Consumer.Offsets.AutoCommit.Enable = true              // 自动提交
// 	// config.Consumer.Offsets.AutoCommit.Interval = 1 * time.Second // 间隔
// 	config.Consumer.Offsets.Initial = sarama.OffsetNewest
// 	//config.Consumer.Offsets.Retry.Max = 3

// 	address := []string{"112.124.13.243:9092"}
// 	consumer, err := sarama.NewConsumer(address, config)
// 	if err != nil {
// 		fmt.Printf("consumer_test create consumer error %s\n", err.Error())
// 		return
// 	}

// 	defer consumer.Close()

// 	partition_consumer, err := consumer.ConsumePartition("test", 0, sarama.OffsetNewest)
// 	if err != nil {
// 		fmt.Printf("try create partition_consumer error %s\n", err.Error())
// 		return
// 	}
// 	defer partition_consumer.Close()

// 	for {
// 		select {
// 		case msg := <-partition_consumer.Messages():
// 			fmt.Printf("msg offset: %d, partition: %d, timestamp: %s, value: %s\n",
// 				msg.Offset, msg.Partition, msg.Timestamp.String(), string(msg.Value))
// 		case err := <-partition_consumer.Errors():
// 			fmt.Printf("err :%s\n", err.Error())
// 		}
// 	}
// }
